package slh.myMapReduce;

import java.io.IOException;
import java.util.TreeMap;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TopRatedMovieReducer extends Reducer<Text, FloatWritable, Text, FloatWritable> {

    private TreeMap<Float, Text> topMovies = new TreeMap<>();

    @Override
    public void reduce(Text key, Iterable<FloatWritable> values, Context context)
            throws IOException, InterruptedException {

        // 由于Mapper已经处理过，这里每个key只有一个value
        for (FloatWritable val : values) {
            float score = val.get();
            // 使用负数实现降序排序
            topMovies.put(-score, new Text(key));

            // 只保留Top3
            if (topMovies.size() > 3) {
                topMovies.remove(topMovies.lastKey());
            }
            break; // 只需要第一个值
        }
    }

    @Override
    protected void cleanup(Context context)
            throws IOException, InterruptedException {

        // 输出Top3电影结果
        for (Float score : topMovies.keySet()) {
            // 拆分出电影标题
            String[] parts = topMovies.get(score).toString().split("\\|");
            String output = parts.length > 1 ? parts[1] : parts[0]; // 使用标题或ID
            context.write(new Text(output), new FloatWritable(-score));
        }
    }
}